package com.atguigu.day09;

import com.atguigu.utils.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<UserBehavior> source = env
                // userId,itemID,categoryId,type,ts
                // 543462,1715,1464116,pv,1511658000
                .readTextFile("D:\\code\\git\\yanzl_PC\\flinktutorial0701\\src\\main\\resources\\UserBehavior.csv")
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] arr = value.split(",");
                        return new UserBehavior(
                                arr[0], arr[1], arr[2], arr[3],
                                Long.parseLong(arr[4]) * 1000L
                        );
                    }
                })
                .filter(r -> r.type.equals("pv"))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);

        Table table = streamTableEnvironment
                .fromDataStream(
                        source,
                        $("userId"),
                        $("itemId"),
                        $("categoryId"),
                        $("type"),
                        $("ts").rowtime()
                );

        streamTableEnvironment.createTemporaryView("userbehavior", table);

        // HOP滑动窗口
        Table result = streamTableEnvironment
                .sqlQuery(
                        "SELECT itemId, COUNT(itemId) as cnt, " +
                                "  HOP_START(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowStart, " +
                                "  HOP_END(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS) as windowEnd" +
                                " FROM userbehavior GROUP BY itemId, HOP(ts, INTERVAL '5' MINUTES, INTERVAL '1' HOURS)"
                );

        streamTableEnvironment.toChangelogStream(result).print();

        env.execute();
    }
}

